package com.shujia.flink.tf

import org.apache.flink.streaming.api.scala._

object Demo6Agg {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")


    /**
      * 取出每个班级年龄最大的学生
      *
      */
    val stuDS: DataStream[(String, String, Int, String, String)] = studentDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(0), split(1), split(2).toInt, split(3), split(4))
    })

    val keyByDS: KeyedStream[(String, String, Int, String, String), String] = stuDS.keyBy(_._5)

    /**
      *
      * sum
      * max
      * min
      * maxBy
      * minBy
      *
      * max 和 maxBy 之间的区别在于 max 返回流中的最大值，但 maxBy 返回具有最大值的键， min 和 minBy 同理。
      *
      *
      *
      */

    val maxDS: DataStream[(String, String, Int, String, String)] = keyByDS.maxBy(2)

    maxDS.print()

    env.execute()

  }

}
